package ds_industry_2025.ds.ds02.T4

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties
/*
    剔除订单信息表与订单详细信息表中用户id与商品id不存在现有的维表中的记录，同时建议多利用缓存并充分考虑并行度来优化代码，达到更快的
    计算效果。

1、根据Hudi的dwd库中相关表或MySQL中shtd_store中相关表（order_detail、sku_info），计算出与用户id为6708的用户所购买相同商品
种类最多的前10位用户（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将10位用户id进行输出，输出格式如下，将结果截
图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
结果格式如下：
-------------------相同种类前10的id结果展示为：--------------------
1,2,901,4,5,21,32,91,14,52
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  todo 导入隐式转换
    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    //  todo 定义读取mysql数据的方法
    def read(tablename:String):DataFrame={
      spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false",tablename,conn)
    }

      val order_info=read("order_info")
    val detail_info=read("order_detail")
    val sku_info=read("sku_info")
    val user_info=read("user_info")

    //  todo 剔除订单表和订单详细表中的user_id和sku_id不存在当前维度表中的数据
    val skus=sku_info.select("id").distinct().withColumnRenamed("id","id_1")
    val users=user_info.select("id").distinct().withColumnRenamed("id","id_1")
    val order=order_info.join(users,users("id_1")===order_info("user_id"),"inner")
    val detail=detail_info.join(skus,skus("id_1")===detail_info("sku_id"),"inner")

    //  todo 把所有用户和购买商品的数据拿到
    val source = order.join(detail, detail("order_id") === order("id"))
      .select("user_id", "sku_id")
      .distinct()

    //  把用户6708购买的商品的种类提取成数组
    //  todo 这里之所以需要把数据类型换成数值类型，是因为字段里面的sku_id也是数值类型，为了数据类型一样
    val user6708_sku:Array[Int] = source.where(col("user_id") === 5241)
      .select("sku_id").distinct()
      .map(r => r(0).toString.toInt)
      .collect()

    user6708_sku.foreach(println)

    //  todo 判断用户购买的商品是否和6708的一样
    //  TODO 为什么在使用groupBy之后不可以在agg聚合函数里面直接使用wihtCOlumn创建字段，因为withColumn只能适用于dataframe
    //  todo 在groupBy之后数据类型就不是dataframe了,而是RelationnalGroupedDataset，我们想要创建字段的话只能在agg里面
    //  todo 使用聚合函数然后as(字段名)
//    val result=source.select("user_id","sku_id")
//      .filter(col("user_id")!==5241)
//      .withColumn(
//      "cos",
//      when(col("sku_id").isin(user6708_sku: _*),lit(1.0)).otherwise(0.0)
//    )
//      .groupBy("user_id")
//      .agg(sum("cos").as("some"))
//      .orderBy(desc("some"),asc("user_id"))
//      .limit(10)
//      .select("user_id")
//      .map(r => r(0).toString)
//      .collect()


    val result = source.select("user_id", "sku_id")
      .filter(col("user_id") !== 5241)
      .withColumn(
        "cos",
        when(col("sku_id").isin(user6708_sku: _*), lit(1.0)).otherwise(0.0)
      )
      .withColumn(
        "some",
        sum("cos").over(Window.partitionBy("user_id"))
        )
      .orderBy(desc("some"), asc("user_id"))
      .limit(10)
      .select("user_id")
      .map(r => r(0).toString)
      .collect()


    println("--------------------种类前10的id展示为:-------------------")
    println(result.mkString(","))












    spark.close()
  }

}
